package com.server.util;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import org.apache.commons.codec.digest.DigestUtils;
import org.jboss.marshalling.Pair;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.StreamMessageId;
import org.redisson.api.stream.StreamAddArgs;
import org.redisson.config.Config;
import redis.clients.jedis.*;
import redis.clients.jedis.args.GeoUnit;
import redis.clients.jedis.commands.JedisCommands;
import redis.clients.jedis.params.BitPosParams;
import redis.clients.jedis.params.GeoAddParams;
import redis.clients.jedis.params.GeoRadiusParam;
import redis.clients.jedis.params.ScanParams;
import redis.clients.jedis.resps.GeoRadiusResponse;
import redis.clients.jedis.resps.ScanResult;

import java.io.File;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * redis工具集
 */
public class RedisUtil {
    /**
     * 基本操作
     */
    private JedisPool jedisPool;
    /**
     * 分布式锁
     */
    private Redisson redisson;
    /**
     * lua脚本所在相对目录
     */
    private static final String scriptDir = "lua_script/";
    /**
     * 在redis中执行lua脚本，保证执行逻辑的原子性，从而免除逻辑中加分布式锁
     */
    private final Map<String, String> redisLuaScript = new HashMap<>();

    private RedisUtil() {

    }

    /**
     * 多个redis实例
     */
    private static final Map<RedisType, RedisUtil> redisInstanceMap = new HashMap<>();

    /**
     * 根据指定类型获取redis连接工具
     */
    public static RedisUtil getInstance(RedisType type) {
        return redisInstanceMap.get(type);
    }

    /**
     * 服务器启动
     * 若不指定redis类型，全部初始化
     */
    public static void init(RedisType... types) {
        RedisType[] initList = types;
        if (types.length == 0) {
            initList = RedisType.values();
        }
        for (RedisType type : initList) {
            if (!type.start) {
                continue;
            }
            RedisUtil instance = new RedisUtil();
            instance.onGameServerStart(type);
            redisInstanceMap.put(type, instance);
        }
    }

    /**
     * 服务器启动
     */
    private void onGameServerStart(RedisType redisType) {
        PropertiesFileUtil conf = PropertiesFileUtil.getInstance(redisType.confName);
        String ip = conf.get("redis.host");
        int port = conf.getInt("redis.port");
        Integer database = conf.getInt("redis.database");
        database = Objects.isNull(database) ? 0 : database;
        String password = conf.get("redis.password");
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        jedisPoolConfig.setMaxTotal(conf.getInt("redis.maxTotal"));
        jedisPoolConfig.setMaxIdle(conf.getInt("redis.maxIdle"));
        jedisPoolConfig.setMinIdle(conf.getInt("redis.minIdle"));
        jedisPoolConfig.setMaxWaitMillis(conf.getInt("redis.maxWaitMillis"));
        jedisPoolConfig.setTestOnBorrow(conf.getBool("redis.testOnBorrow"));
        jedisPoolConfig.setTestOnReturn(conf.getBool("redis.testOnReturn"));
        jedisPoolConfig.setTestWhileIdle(conf.getBool("redis.testWhileIdle"));
        //
        Config config = new Config();
        //
        if (password != null && !"".equals(password)) {
            // redis 设置了密码
            jedisPool = new JedisPool(jedisPoolConfig, ip, port, 10000, password, database);
            config.useSingleServer().setAddress("redis://" + ip + ":" + port).setPassword(password);
        } else {
            // redis 未设置密码
            jedisPool = new JedisPool(jedisPoolConfig, ip, port, 10000, null, database);
            config.useSingleServer().setAddress("redis://" + ip + ":" + port);
        }
        //
        redisson = (Redisson) Redisson.create(config);
        //加载lua脚本至redis
        if (redisType.execLua) {
            File dir = new File(scriptDir);
            File[] files = dir.listFiles();
            if (null == files || files.length == 0) {
                LogUtil.game.warn("lua file not found!  " + scriptDir);
            } else {
                Arrays.stream(files).filter(f -> !f.isDirectory() && f.getName().endsWith(".lua")).forEach(f -> redisLuaScript.put(f.getName().substring(0, f.getName().length() - 4), scriptDir + f.getName()));
            }
            //
            var it = redisLuaScript.entrySet().iterator();
            while (it.hasNext()) {
                var entry = it.next();
                String scriptName = entry.getValue();
                String scriptStr = FileUtil.readAllFileText(scriptName);
                String sha1 = DigestUtils.sha1Hex(scriptStr);
                entry.setValue(sha1);
                if (scriptLuaExists(sha1)) {
                    continue;
                }
                if (!scriptLuaLoad(scriptStr).equals(sha1)) {
                    LogUtil.game.error("RedisManager init scriptLua err:" + scriptName, new RuntimeException("RedisManager init scriptLua err:" + scriptName));
                }
            }
        }
    }

    public static void close() {
        for (RedisUtil redis : redisInstanceMap.values()) {
            redis.jedisPool.close();
            redis.redisson.shutdown();
        }
    }

    private Jedis getJedis() {
        return jedisPool.getResource();
    }

    private void call(Consumer<JedisCommands> consumer) {
        try (Jedis jedis = getJedis()) {
            consumer.accept(jedis);
        } catch (Exception e) {
            LogUtil.game.error("redis ex", e);
        }
    }

    private <R> R call(Function<JedisCommands, R> function, R defaultReturn) {
        try (Jedis jedis = getJedis()) {
            return function.apply(jedis);
        } catch (Exception e) {
            LogUtil.game.error("redis ex", e);
        }
        return defaultReturn;
    }

    @SuppressWarnings("unchecked")
    private <T> List<T> toList(Collection<String> value, Class<T> clazz) {
        if (value != null) {
            if (clazz == String.class) {
                return (List<T>) new ArrayList<>(value);
            }
            List<T> newValue = Lists.newArrayList();
            for (String temp : value) {
                newValue.add(JSON.parseObject(temp, clazz));
            }
            return newValue;
        }
        return null;
    }

    /***Lua操作**********************************************************************************/

    public boolean scriptLuaExists(String sha1) {
        try (Jedis jedis = getJedis()) {
            return jedis.scriptExists(sha1);
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
        return false;
    }

    public String scriptLuaLoad(String luaScript) {
        try (Jedis jedis = getJedis()) {
            return jedis.scriptLoad(luaScript);
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
        return null;
    }

    /**
     * 执行lua脚本
     */
    public <T> List<T> scriptLua(String scriptName, List<String> keys, List<String> args) {
        String scriptSha = redisLuaScript.get(scriptName);
        List<T> list = new ArrayList<>();
        try (Jedis jedis = getJedis()) {
            //传入的参数
            // /usr/local/bin/redis-cli -h 127.0.0.1 -p 6379 --eval /usr/local/luadir/03.lua name age , baihezhuo
            List<T> ret = (List<T>) jedis.evalsha(scriptSha, keys, args);
            if (Objects.nonNull(ret) && ret.size() > 0) {
                list.addAll(ret);
            }
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
        return list;
    }

    /***redission操作**********************************************************************************/

    /**
     * redission 分布式锁
     *
     * @param lockName  锁名
     * @param waitTime  重试时间
     * @param leaseTime 锁过期时间
     * @param timeUnit  自定义时间单位
     */
    public boolean tryLock(String lockName, long waitTime, long leaseTime, TimeUnit timeUnit) {
        try {
            RLock rLock = redisson.getLock(lockName);
            return rLock.tryLock(waitTime, leaseTime, timeUnit);
        } catch (Exception e) {
            LogUtil.game.error("redission lock error with waitTime", e);
        }
        return false;
    }

    /**
     * redission 分布式锁
     * 释放锁
     */
    public void unlock(String lockName) {
        try {
            RLock rLock = redisson.getLock(lockName);
            if (Objects.nonNull(rLock)) {
                rLock.unlock();
            }
        } catch (Exception e) {
            LogUtil.game.error("redission releaseLock error" + lockName, e);
        }
    }

    /**
     * 布隆过滤器
     */
    public boolean bloomFilterAdd(String name, Object object) {
        try {
            return redisson.getBloomFilter(name).add(object);
        } catch (Exception e) {
            LogUtil.game.error("BloomFilter", e);
        }
        return false;
    }

    public boolean bloomFilterContains(String name, Object object) {
        try {
            return redisson.getBloomFilter(name).contains(object);
        } catch (Exception e) {
            LogUtil.game.error("BloomFilter", e);
        }
        return false;
    }

    public boolean bloomFilterDelete(String name) {
        try {
            return redisson.getBloomFilter(name).delete();
        } catch (Exception e) {
            LogUtil.game.error("BloomFilter", e);
        }
        return false;
    }

    /***基本操作 pipeline方式**********************************************************************************/
    public List<Object> get_Pipeline(List<String> keys) {
        try {
            Jedis jedis = getJedis();
            Pipeline pipelined = jedis.pipelined();
            for (String key : keys) {
                pipelined.get(key);
            }
            List<Object> list = pipelined.syncAndReturnAll();
            return list;
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
        return new ArrayList<>();
    }

    public void del_Pipeline(List<String> keys) {
        try {
            Jedis jedis = getJedis();
            Pipeline pipelined = jedis.pipelined();
            for (String key : keys) {
                pipelined.del(key);
            }
            pipelined.sync();
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
    }

    public <T> void set_Pipeline(List<String> keys, List<T> objects) {
        assert keys.size() == objects.size();
        try {
            Jedis jedis = getJedis();
            Pipeline pipelined = jedis.pipelined();
            for (int i = 0; i < keys.size(); i++) {
                pipelined.set(keys.get(i), JSON.toJSONString(objects.get(i)));
            }
            pipelined.sync();
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
    }

    public List<Object> hmget_Pipeline(List<String> keys, String... fields) {
        try {
            Jedis jedis = getJedis();
            Pipeline pipelined = jedis.pipelined();
            for (String key : keys) {
                pipelined.hmget(key, fields);
            }
            List<Object> list = pipelined.syncAndReturnAll();
            return list;
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
        return new ArrayList<>();
    }

    public <T> void hmset_Pipeline(List<String> keys, List<Map<String, T>> hashList) {
        assert keys.size() == hashList.size();
        try {
            Jedis jedis = getJedis();
            Pipeline pipelined = jedis.pipelined();
            for (int i = 0; i < keys.size(); i++) {
                pipelined.hmset(keys.get(i), (Map<String, String>) hashList.get(i));
            }
            pipelined.sync();
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
    }

    public List<Object> hgetAll_Pipeline(List<String> keys) {
        try {
            Jedis jedis = getJedis();
            Pipeline pipelined = jedis.pipelined();
            for (String key : keys) {
                pipelined.hgetAll(key);
            }
            List<Object> list = pipelined.syncAndReturnAll();
            return list;
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
        return new ArrayList<>();
    }

    public void srem_Pipeline(String key, List<String> values) {
        try {
            Jedis jedis = getJedis();
            Pipeline pipelined = jedis.pipelined();
            for (String value : values) {
                pipelined.srem(key, value);
            }
            pipelined.syncAndReturnAll();
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
    }

    public void zadd_Pipeline(String key, Map<String, Double> memberScoreMap) {
        try {
            Jedis jedis = getJedis();
            Pipeline pipelined = jedis.pipelined();
            for (String member : memberScoreMap.keySet()) {
                pipelined.zadd(key, memberScoreMap.getOrDefault(member, 0D), member);
            }
            pipelined.syncAndReturnAll();
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
    }

    public Map<String, Double> zscore_Pipeline(String key, List<String> memberList) {
        Map<String, Double> map = new HashMap<>();
        try {
            Jedis jedis = getJedis();
            Pipeline pipelined = jedis.pipelined();
            for (String member : memberList) {
                pipelined.zscore(key, member);
            }
            //
            List<Object> resultList = pipelined.syncAndReturnAll();
            for (int i = 0; i < memberList.size(); i++) {
                map.put(memberList.get(i), Objects.isNull(resultList.get(i)) ? 0D : (Double) resultList.get(i));
            }
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
        return map;
    }

    /***基本操作 尽量用 call(...)来统一调用**********************************************************************************/
    @SuppressWarnings("unchecked")
    public <T> T get(String key, Class<T> clazz) {
        String value = call(jedis -> jedis.get(key), null);
        if (value == null) {
            return null;
        }
        if (clazz == String.class) {
            return (T) value;
        }
        return JSON.parseObject(value, clazz);
    }

    public void set(String key, String value) {
        set(key, value, 0);
    }

    public void set(String key, Object value) {
        set(key, value, 0);
    }

    public void set(String key, Object value, int time) {
        set(key, JSON.toJSONString(value), time);
    }

    public void set(String key, String value, int time) {
        call(jedis -> {
            jedis.set(key, value);
            if (time > 0) {
                jedis.expire(key, time);
            }
        });
    }

    public long del(String key) {
        return call(jedis -> jedis.del(key), 0L);
    }

    public void expire(String key, int seconds) {
        call(jedis -> {
            jedis.expire(key, seconds);
        });
    }

    public long zadd(String key, double score, String member) {
        return call(jedis -> jedis.zadd(key, score, member), -1L);
    }

    public long zcard(String key) {
        return call(jedis -> jedis.zcard(key), 0L);
    }

    public long zrem(String key, String value) {
        return call(jedis -> jedis.zrem(key, value), -1L);
    }

    public void zrem(String key, List<String> values) {
        call(jedis -> jedis.zrem(key, values.toArray(new String[0])));
    }

    /**
     * 通过排名移除成员
     */
    public void zRemByRank(String key, int start, int end) {
        call(jedis -> jedis.zremrangeByRank(key, start, end));
    }

    /**
     * 获取集合数量
     */
    public long scard(String key) {
        return call(jedis -> jedis.scard(key), 0L);
    }

    /**
     * 从列表中获取指定返回的元素 start 和 end
     * 偏移量都是基于0的下标，即list的第一个元素下标是0（list的表头），第二个元素下标是1，以此类推。
     * 偏移量也可以是负数，表示偏移量是从list尾部开始计数。 例如， -1 表示列表的最后一个元素，-2 是倒数第二个，以此类推。
     * 从小到大排行
     */
    public <T> List<T> zrange(String key, int start, int end, Class<T> clazz) {
        List<String> value = call(jedis -> jedis.zrange(key, start, end), Collections.EMPTY_LIST);
        return toList(value, clazz);
    }

    /**
     * 根据分数区间获取排名列表(按score从小到大排列)
     */
    public List<String> zrangeByScore(String key, double startValue, double endValue) {
        return call(jedis -> jedis.zrangeByScore(key, startValue, endValue), Collections.EMPTY_LIST);
    }

    /**
     * 获取指定成员member的排名 (按score从小到大排列)
     * 排名以0为底,也就是说score值最小的成员排名为0
     */
    public Long zrank(String key, String member) {
        return call(jedis -> jedis.zrank(key, member), -1L);
    }

    /**
     * 返回指定区间分数的成员数量
     */
    public Long zcount(String key, long min, long max) {
        return call(jedis -> jedis.zcount(key, min, max), 0L);
    }

    /**
     * 从列表中获取指定返回的元素 start 和 end
     * 偏移量都是基于0的下标，即list的第一个元素下标是0（list的表头），第二个元素下标是1，以此类推。
     * 偏移量也可以是负数，表示偏移量是从list尾部开始计数。 例如， -1 表示列表的最后一个元素，-2 是倒数第二个，以此类推。
     * 从大到小排行
     */
    public <T> List<T> zrevrange(String key, int start, int end, Class<T> clazz) {
        List<String> value = call(jedis -> jedis.zrevrange(key, start, end), Collections.EMPTY_LIST);
        return toList(value, clazz);
    }

    /**
     * 获取指定成员member的排名 (按score从大到小排列)
     * 排名以0为底,也就是说score值最小的成员排名为0
     */
    public Long zrevrank(String key, String member) {
        return call(jedis -> jedis.zrevrank(key, member), -1L);
    }

    /**
     * 获取指定成员的分数
     */
    public Double zscore(String key, String member) {
        return call(jedis -> jedis.zscore(key, member), 0D);
    }

    /**
     * 获取指定成员列表的分数
     */
    public Map<String, Double> zscore_list_pipelined(String key, List<String> memberList) {
        Map<String, Double> map = new HashMap<>();
        try (Jedis jedis = getJedis(); Pipeline pipelined = jedis.pipelined()) {
            for (String member : memberList) {
                pipelined.zscore(key, member);
            }
            //
            List<Object> resultList = pipelined.syncAndReturnAll();
            for (int i = 0; i < memberList.size(); i++) {
                map.put(memberList.get(i), Objects.isNull(resultList.get(i)) ? 0D : (Double) resultList.get(i));
            }
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
        return map;
    }


    /**
     * ursor：游标，当次遍历的起始位置
     * count：返回数据条数，但只是一个hint(暗示)，具体返回条数可多可少。
     * 扫描指定的set类型的key
     */
    public <T> List<T> sscan(String key, Class<T> clazz, int cursor, int count) {
        List<String> list = call(jedis -> jedis.sscan(key, Integer.toString(cursor), new ScanParams().count(count)).getResult(), Collections.EMPTY_LIST);
        return toList(list, clazz);
    }

    /**
     * cursor：游标，当次遍历的起始位置
     * pattern：与Keys命令中的patterns相同，支持通配符匹配
     * count：返回数据条数，但只是一个hint(暗示)，具体返回条数可多可少。
     */
    public <T> Pair<String, List<T>> scan(String cursor, Class<T> clazz, String pattern, int count) {
        try (Jedis jedis = getJedis()) {
            ScanResult<String> result = jedis.scan(cursor, new ScanParams().match(pattern).count(count));
            return new Pair<>(result.getCursor(), toList(result.getResult(), clazz));
        } catch (Exception e) {
            LogUtil.game.error("redis ex", e);
        }
        return null;
    }

    /**
     * Bitmaps位图
     * 可按位存取，也可普通get set
     * Bitmaps底层就是字符串（key-value）byte数组
     * offset：0~2^32
     */
    public boolean setbit(String key, long offset, boolean value) {
        return call(jedis -> jedis.setbit(key, offset, value), false);
    }

    public boolean getbit(String key, long offset) {
        return call(jedis -> jedis.getbit(key, offset), false);
    }

    /**
     * 计算给定字符串中，被设置为1的bit位的数量
     * start和end是字节索引，因此每增加1 代表的是增加一个字符，也就是8位，所以位的查询范围必须是8的倍数
     * -1代表最后一个字节，-2代表倒是第二个字节。
     */
    public long bitcount(String key, long start, long end) {
        return call(jedis -> jedis.bitcount(key, start, end), 0L);
    }

    public long bitpos(String key, long start, long end) {
        return call(jedis -> jedis.bitpos(key, true, new BitPosParams(start, end)), -1L);
    }

    /**
     * HyperLogLog结构
     * 适用UV统计 标准误差0.81%
     */
    public long pfadd(String key, String... elements) {
        return call(jedis -> jedis.pfadd(key, elements), 0L);
    }

    public long pfcount(String key) {
        return call(jedis -> jedis.pfcount(key), 0L);
    }

    /**
     * Geo spatial 地理位置GEO模块
     * 对两极不支持
     * longitude有效经度介于-180°~180°之间
     * latitude有效维度介于-85.05112878°至85.05112878°之间
     */
    public long geoadd(String key, double longitude, double latitude, String member) {
        GeoAddParams params = new GeoAddParams();
        /**
         * XX:只更新已经存在的元素。不要添加元素。
         * NX:不要更新已经存在的元素。总是添加新元素。
         * CH:根据new的数目修改返回值添加的元素，加上改变的元素总数(CH是一个changed的缩写)。已更改的元素是添加的新元素和已存在的元素，这些元素的坐标已更新。所以在命令行中指定的元素具有与它们相同的分数
         * 过去的都不算。
         * 注意:通常，GEOADD的返回值只计算添加的新元素的数量。注意:XX和NX选项互斥。
         */
        Map<String, GeoCoordinate> memberCoordinateMap = new HashMap<>();
        memberCoordinateMap.put(member, new GeoCoordinate(longitude, latitude));
        return call(jedis -> jedis.geoadd(key, params, memberCoordinateMap), 0L);
    }

    /**
     * 6.2版本之后弃用，给定圆心和半径，查找圆形范围内的数据
     *
     * @param radius
     * @param unit   GeoUnit.KM  radius 单位
     * @param param  模式 GeoRadiusParam.geoRadiusParam().withDist()
     */
    public List<GeoRadiusResponse> georadius(String key, double longitude, double latitude, double radius, GeoUnit unit, GeoRadiusParam param) {
        try {
            Jedis jedis = getJedis();
            List<GeoRadiusResponse> result = jedis.georadius(key, longitude, latitude, radius, unit, param);
            return result;
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
        return null;
    }

    /**
     * 6.2版本之后弃用，指定GEO集合中的某个成员为中心，圆形范围查找
     */
    public List<GeoRadiusResponse> georadiusByMember(String key, String member, double radius, GeoUnit unit, GeoRadiusParam param) {
        try {
            Jedis jedis = getJedis();
            List<GeoRadiusResponse> result = jedis.georadiusByMember(key, member, radius, unit, param);
            return result;
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
        return null;
    }

    /**
     * 【6.2版本新增】，可设定按矩形范围查询还是圆形范围查询
     *
     * @param longitude
     * @param latitude
     * @param width     宽
     * @param height    高
     * @param unit
     */
    public List<GeoRadiusResponse> geosearch_rectangle(String key, double longitude, double latitude, double width, double height, GeoUnit unit) {
        try {
            Jedis jedis = getJedis();
            GeoCoordinate g = new GeoCoordinate(longitude, latitude);
            List<GeoRadiusResponse> result = jedis.geosearch(key, g, width, height, unit);
            return result;
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
        return null;
    }

    /**
     * 6.2版本之后弃用，指定GEO集合中的某个成员为中心，圆形范围查找
     */
    public List<GeoRadiusResponse> geosearch_radius(String key, double longitude, double latitude, double radius, GeoUnit unit) {
        try {
            Jedis jedis = getJedis();
            GeoCoordinate g = new GeoCoordinate(longitude, latitude);
            List<GeoRadiusResponse> result = jedis.geosearch(key, g, radius, unit);
            return result;
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
        return null;
    }

    public List<GeoRadiusResponse> geosearch_member_radius(String key, String member, double radius, GeoUnit unit) {
        try {
            Jedis jedis = getJedis();
            List<GeoRadiusResponse> result = jedis.geosearch(key, member, radius, unit);
            return result;
        } catch (Exception ex) {
            LogUtil.game.error("RedisUtil err", ex);
        }
        return null;
    }
}
